# main.py

import asyncio
from client.session import create_session
from router import route_message
from mcp.types import TextContent, CreateMessageRequestParams
from mcp.shared.context import RequestContext
from mcp import ClientSession

# 任务处理函数
async def process_input(
    context: RequestContext[ClientSession, None],
    params: CreateMessageRequestParams
):
    # 获取用户输入内容（如邮件内容文本）
    message_text = params.messages[0].content.text.strip()
    
    # 通过路由器进行意图判断与任务分发
    result = await route_message(context, message_text)
    
    return result

# 启动主入口函数
async def main():
    session = await create_session(callback=process_input)
    try:
        await session.initialize()
        print("[系统已启动] 输入任意邮件内容开始处理，输入 quit 退出：")
        while True:
            query = input("邮件内容> ").strip()
            if query.lower() == "quit":
                break
            await session.inject_text(query)
    finally:
        await session.exit_stack.aclose()

if __name__ == "__main__":
    asyncio.run(main())
    
    
    
    
# router.py

from mcp.shared.context import RequestContext
from mcp import ClientSession
from mcp.types import TextContent, CreateMessageResult
import re

# 简易意图识别函数
def classify_intent(text: str) -> str:
    if "总结" in text or "概括" in text:
        return "summarizer"
    elif "归档" in text or "标签" in text:
        return "archiver"
    elif "回复" in text or "答复" in text:
        return "reply_generator"
    elif "分类" in text or "是什么类型" in text:
        return "classifier"
    else:
        return "mail_parser"

# 路由器主函数
async def route_message(
    context: RequestContext[ClientSession, None],
    text: str
) -> CreateMessageResult:
    
    intent = classify_intent(text)
    print(f"[路由判断] 当前意图：{intent}")

    response = await context.session.call_tool(
        tool_name=intent,
        tool_input={"text": text}
    )

    return CreateMessageResult(
        role="assistant",
        content=TextContent(type="text", text=response.content[0].text),
        model="mcp-router",
        stopReason="endTurn"
    )
    
    
# context_types.py

from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime

# 通用邮件结构定义
class MailMeta(BaseModel):
    sender: str = Field(..., description="发件人邮箱地址")
    receiver: str = Field(..., description="收件人邮箱地址")
    subject: str = Field(..., description="邮件主题")
    timestamp: datetime = Field(..., description="邮件接收时间")

class MailBody(BaseModel):
    plain_text: str = Field(..., description="邮件正文内容")
    html: Optional[str] = Field(None, description="HTML格式正文（如有）")

class MailAttachment(BaseModel):
    filename: str = Field(..., description="附件名称")
    filetype: str = Field(..., description="文件类型，如pdf、jpg")
    filesize_kb: int = Field(..., description="文件大小KB")

class MailContext(BaseModel):
    meta: MailMeta
    body: MailBody
    attachments: Optional[List[MailAttachment]] = Field(default_factory=list)

# 分类结构
class ClassificationResult(BaseModel):
    category: str = Field(..., description="邮件类别：如事务、系统、广告")
    confidence: float = Field(..., description="分类置信度")

# 摘要结构
class SummaryResult(BaseModel):
    summary: str = Field(..., description="生成的摘要内容")

# 回复建议结构
class ReplyCandidate(BaseModel):
    reply_text: str = Field(..., description="建议回复内容")
    intent: Optional[str] = Field(None, description="意图类型，如确认、拒绝、需跟进")

# 归档结构
class ArchiveMetadata(BaseModel):
    folder: str = Field(..., description="归档文件夹名称")
    tags: List[str] = Field(..., description="归档标签")
    
    
    

# 模板：摘要生成
SUMMARY_PROMPT_TEMPLATE = """
请对以下邮件内容生成一段简洁明了的中文摘要，控制在100字以内：

---
{content}
---
摘要：
"""

# 模板：分类任务
CLASSIFY_PROMPT_TEMPLATE = """
以下是一封电子邮件内容，请判断其属于哪一类（事务、广告、系统通知、社交）：

---
{content}
---
邮件类别：
"""

# 模板：邮件回复构造
REPLY_PROMPT_TEMPLATE = """
请基于以下邮件内容，构造一条合适的回复语句。语气需保持正式和礼貌：

---
{content}
---
建议回复：
"""

# 模板：归档与标签标注
ARCHIVE_PROMPT_TEMPLATE = """
请根据以下邮件内容，建议一个合适的归档文件夹与标签集合，用于邮件分类与后续检索：

---
{content}
---
建议归档路径与标签：
"""




from mcp.server import tool
from pydantic import BaseModel
from context_types import MailContext, MailMeta, MailBody
from datetime import datetime

class MailParserInput(BaseModel):
    raw_text: str

@tool()
async def mail_parser(input: MailParserInput) -> MailContext:
    # 伪解析逻辑，可拓展为正则提取或结构化邮件头解析
    return MailContext(
        meta=MailMeta(
            sender="user@example.com",
            receiver="bot@example.com",
            subject="测试邮件",
            timestamp=datetime.utcnow()
        ),
        body=MailBody(
            plain_text=input.raw_text
        )
    )
    
    
    
from pydantic import BaseModel
from context_types import ClassificationResult

class ClassifierInput(BaseModel):
    text: str

@tool()
async def classifier(input: ClassifierInput) -> ClassificationResult:
    content = input.text.lower()
    if "会议" in content or "汇报" in content:
        category = "事务通知"
    elif "验证码" in content or "系统" in content:
        category = "系统信息"
    elif "促销" in content or "优惠" in content:
        category = "广告推广"
    else:
        category = "社交沟通"
    return ClassificationResult(category=category, confidence=0.9)



from mcp.server import tool
from pydantic import BaseModel
from context_types import ReplyCandidate

class ReplyInput(BaseModel):
    text: str

@tool()
async def reply_generator(input: ReplyInput) -> ReplyCandidate:
    reply = "您好，邮件已收到，将尽快处理，感谢联系。"
    return ReplyCandidate(reply_text=reply, intent="确认")





from mcp.server import tool
from pydantic import BaseModel
from context_types import ArchiveMetadata

class ArchiveInput(BaseModel):
    text: str

@tool()
async def archiver(input: ArchiveInput) -> ArchiveMetadata:
    folder = "事务归档"
    tags = ["项目", "总结", "处理完毕"]
    return ArchiveMetadata(folder=folder, tags=tags)



# server/app.py

from mcp.server import FastMCP

# 导入各类MCP工具
from tools.mail_parser import mail_parser
from tools.classifier import classifier
from tools.summarizer import summarizer
from tools.reply_generator import reply_generator
from tools.archiver import archiver

# 创建 FastMCP 实例（服务名称：mail-agent）
app = FastMCP("mail-agent")

# 注册工具
app.register_tool(mail_parser)
app.register_tool(classifier)
app.register_tool(summarizer)
app.register_tool(reply_generator)
app.register_tool(archiver)





import asyncio
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters
from mcp.types import CreateMessageRequestParams, CreateMessageResult, TextContent
from mcp.shared.context import RequestContext

# FastMCP服务端文件路径
server_params = StdioServerParameters(
    command="uv", args=["run", "server/app.py"]
)

# 回调函数处理用户输入
async def default_callback(
    context: RequestContext[ClientSession, None],
    params: CreateMessageRequestParams
) -> CreateMessageResult:
    # 简单应答
    text = params.messages[0].content.text
    return CreateMessageResult(
        role="assistant",
        content=TextContent(type="text", text=f"收到：{text}"),
        model="echo-agent",
        stopReason="endTurn"
    )

# 外部调用方法：创建会话
async def create_session(callback=default_callback) -> ClientSession:
    stdio, write = await stdio_client(server_params).__aenter__()
    session = ClientSession(stdio, write, sampling_callback=callback)
    return session



# state_machine.py

from enum import Enum, auto

class TaskState(Enum):
    INIT = auto()
    PARSED = auto()
    CLASSIFIED = auto()
    SUMMARIZED = auto()
    REPLIED = auto()
    ARCHIVED = auto()
    COMPLETED = auto()

class StateMachine:
    def __init__(self):
        self.state = TaskState.INIT

    def next(self):
        if self.state == TaskState.INIT:
            self.state = TaskState.PARSED
        elif self.state == TaskState.PARSED:
            self.state = TaskState.CLASSIFIED
        elif self.state == TaskState.CLASSIFIED:
            self.state = TaskState.SUMMARIZED
        elif self.state == TaskState.SUMMARIZED:
            self.state = TaskState.REPLIED
        elif self.state == TaskState.REPLIED:
            self.state = TaskState.ARCHIVED
        elif self.state == TaskState.ARCHIVED:
            self.state = TaskState.COMPLETED
        return self.state

    def is_terminal(self):
        return self.state == TaskState.COMPLETED

    def reset(self):
        self.state = TaskState.INIT




# context_chain.py

from typing import List, Dict
from context_types import MailContext
from copy import deepcopy

class ContextChain:
    def __init__(self):
        self._chain: List[Dict] = []
        self._kv_cache: Dict[str, any] = {}
        self._snapshot = None

    def add_context(self, step_name: str, data: any):
        self._chain.append({"step": step_name, "data": deepcopy(data)})
        self._kv_cache[step_name] = data

    def get_context(self, step_name: str):
        return self._kv_cache.get(step_name)

    def get_all_steps(self):
        return [entry["step"] for entry in self._chain]

    def latest(self):
        return self._chain[-1] if self._chain else None

    def snapshot(self):
        self._snapshot = deepcopy(self._chain)

    def restore(self):
        if self._snapshot:
            self._chain = deepcopy(self._snapshot)
            self._kv_cache = {item["step"]: item["data"] for item in self._chain}



import logging
import os
from datetime import datetime

# 日志输出路径
LOG_DIR = "logs"
LOG_FILE = os.path.join(LOG_DIR, f"mcp_mailagent_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log")

# 确保日志目录存在
os.makedirs(LOG_DIR, exist_ok=True)

# 初始化日志器
logger = logging.getLogger("mcp-mailagent")
logger.setLevel(logging.DEBUG)

# 文件日志处理器
file_handler = logging.FileHandler(LOG_FILE, encoding="utf-8")
file_handler.setLevel(logging.DEBUG)

# 控制台日志处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)

# 格式定义
formatter = logging.Formatter(
    fmt="[%(asctime)s][%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
)

file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)

# 注册处理器
logger.addHandler(file_handler)
logger.addHandler(console_handler)




# debug/inspector.py

import asyncio
from client.session import create_session
from mcp.types import CreateMessageRequestParams, CreateMessageResult, TextContent
from mcp.shared.context import RequestContext
from mcp import ClientSession
from debug.logger import logger

# 调试回调函数
async def debug_callback(
    context: RequestContext[ClientSession, None],
    params: CreateMessageRequestParams
) -> CreateMessageResult:
    text = params.messages[0].content.text.strip()
    logger.info(f"[输入内容] {text}")

    # 手动指定任务意图进行测试（默认使用summarizer）
    try:
        response = await context.session.call_tool("summarizer", {"text": text})
        logger.info(f"[调用成功] tool=summarizer → {response.content[0].text}")
        return CreateMessageResult(
            role="assistant",
            content=TextContent(type="text", text=response.content[0].text),
            model="debug-agent",
            stopReason="endTurn"
        )
    except Exception as e:
        logger.error(f"[调用失败] {e}")
        return CreateMessageResult(
            role="assistant",
            content=TextContent(type="text", text="处理失败，请检查日志"),
            model="debug-agent",
            stopReason="error"
        )

# 启动调试会话
async def debug_main():
    session = await create_session(callback=debug_callback)
    try:
        await session.initialize()
        print("[调试模式] 输入任意内容触发 summarizer 工具调用，输入 quit 退出")
        while True:
            user_input = input("邮件内容> ").strip()
            if user_input.lower() == "quit":
                break
            await session.inject_text(user_input)
    finally:
        await session.exit_stack.aclose()

if __name__ == "__main__":
    asyncio.run(debug_main())





from pydantic import BaseSettings, Field
from typing import Optional

class AppSettings(BaseSettings):
    # Deepseek Chat 接口相关
    deepseek_api_key: str = Field(..., env="DEEPSEEK_API_KEY")
    deepseek_base_url: str = Field("https://api.deepseek.com", env="DEEPSEEK_BASE_URL")
    deepseek_model: str = Field("deepseek-chat", env="DEEPSEEK_MODEL")

    # 日志等级
    log_level: str = Field("INFO", env="LOG_LEVEL")

    # Prompt路径（如有）
    prompt_path: Optional[str] = Field("prompt_templates/", env="PROMPT_PATH")

    # 上下文最大长度限制
    max_context_length: int = Field(4096, env="MAX_CONTEXT_LENGTH")

    # 是否启用调试模式
    debug_mode: bool = Field(False, env="DEBUG_MODE")

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

# 实例化全局配置对象
settings = AppSettings()




# .env（注意：该文件不应纳入版本控制）

DEEPSEEK_API_KEY=sk-xxx-your-deepseek-key
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-chat

LOG_LEVEL=INFO
PROMPT_PATH=prompt_templates/
MAX_CONTEXT_LENGTH=4096
DEBUG_MODE=False




from fastapi.responses import HTMLResponse, JSONResponse
from pydantic import BaseModel, Field
from typing import List, Dict, Optional
import uvicorn
import time

############################
#        MCP MODELS
############################

class MessageContent(BaseModel):
    text: str = Field(..., description="用户输入的文本内容")

class MCPMessage(BaseModel):
    role: str = Field(..., description="消息角色，如user、assistant等")
    content: MessageContent

class MCPRequest(BaseModel):
    model: str = "deepseek-chat"
    messages: List[MCPMessage] = Field(..., description="消息列表")
    metadata: Dict[str, str] = Field(default_factory=dict, description="可选元信息")
    config: Dict[str, any] = Field(default_factory=dict, description="推理配置")

class MCPResponse(BaseModel):
    status: str = "success"
    outputs: List[str] = Field(default_factory=list, description="模型响应内容")
    metadata: Dict[str, str] = Field(default_factory=dict, description="可选的附加元信息")

############################
#      FASTAPI APP
############################

app = FastAPI()

# 模拟服务器端的MCP上下文管理（简化，真实环境可整合状态机与工具链）
session_contexts: Dict[str, List[str]] = {}

@app.get("/")
async def index():
    """
    返回简单的HTML界面，以演示前端如何与MCP接口进行通信
    """
    html_content = """
    <html>
      <head>
        <title>MCP - Deepseek Chat Demo</title>
      </head>
      <body>
        <h2>前端与MCP接口通信示例</h2>
        <div>
          <label>输入文本：</label>
          <input type="text" id="userInput" style="width:300px;" />
          <button onclick="sendRequest()">发送</button>
        </div>
        <pre id="responseArea" style="margin-top:1em; background:#eee; padding:1em;"></pre>

        <script>
          async function sendRequest() {
            const inputVal = document.getElementById('userInput').value;
            const requestPayload = {
              "model": "deepseek-chat",
              "messages": [
                {"role": "user", "content": {"text": inputVal}}
              ],
              "metadata": {"session_id":"demo-session"},
              "config": {"stream": false}
            };
            const res = await fetch("/mcp_chat", {
              method: "POST",
              headers: {
                "Content-Type": "application/json"
              },
              body: JSON.stringify(requestPayload)
            });
            const jsonResp = await res.json();
            document.getElementById('responseArea').innerText = JSON.stringify(jsonResp, null, 2);
          }
        </script>
      </body>
    </html>
    """
    return HTMLResponse(html_content)

@app.post("/mcp_chat")
async def mcp_chat_endpoint(request: MCPRequest) -> MCPResponse:
    """
    MCP接口，用于接收前端发送的JSON请求，并返回对话响应
    """
    # 解析会话ID
    session_id = request.metadata.get("session_id", "unknown")
    if session_id not in session_contexts:
        session_contexts[session_id] = []

    # 提取用户内容
    user_msg = request.messages[-1].content.text
    session_contexts[session_id].append(f"user: {user_msg}")

    # 简易逻辑：将输入文本逆序作为响应
    reversed_text = user_msg[::-1]
    time.sleep(1)  # 模拟服务器处理延迟
    session_contexts[session_id].append(f"assistant: {reversed_text}")

    # 构造返回
    return MCPResponse(
        status="success",
        outputs=[reversed_text],
        metadata={"context_length": str(len(session_contexts[session_id]))}
    )


############################
#     START UVICORN
############################

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)







































































































































